package com.software.process

import com.software.util.DBTools
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.functions.expr
import org.apache.spark.sql.{DataFrame, SparkSession}

//全球气温的统计分析
object GlobalTemperatureProcess extends Serializable {

  Logger.getLogger("org").setLevel(Level.ERROR)

  def main(args: Array[String]): Unit = {
    //编码阶段创建的SparkSession对象
    val sparkSession: SparkSession =
      SparkSession.builder().master("local[*]")
        .appName("GlobalTemperatureProcess")
        .config("spark.testing.memory", "2147480000")
        .getOrCreate()


    //部署运行阶段创建的SparkSession对象
    /*val sparkSession: SparkSession =
      SparkSession.builder()
        .appName("GlobalTemperatureProcess")
        .getOrCreate()*/
    //2.加载csv文件（数据源）

    val inputWebfile = "input/countryTemperatures.csv"
    var weatherData: DataFrame = sparkSession.read.format("csv")
      .option("header", true)
      .option("multiLine", true)
      .load(inputWebfile)

    //正则表达式的替换
    weatherData = regexpReplace(weatherData, "dt", "/", "-")
    weatherData.show(5)

    //3.注册临时表
    weatherData.createOrReplaceTempView("tbl_globalCountryTemperatures")
    //3.1统计2008-2012年全球的平均地表温度
    var result: DataFrame =
      sparkSession.sql(
        """
          |select year(dt) year ,round(avg(AverageTemperature),2) average,Country
          |from tbl_globalCountryTemperatures
          |where AverageTemperature !='null' and dt!=""
          |and year(dt) between 2008 and 2012
          |group by Country,year(dt)
          |order by average desc
          | """.stripMargin)

    result.show(false)
    DBTools.WriteMySql("country_2008_2012_temperatures", result)
    //3.2 统计中国每年每季度的平均气温（1836-2012年）
    val avgresult: DataFrame =
      sparkSession.sql(
        """
          |SELECT year(dt) year,round(AVG(averageTemperature),3) AS averageTemperature,CEIL(MONTH(dt)/3) AS season
          |FROM  tbl_globalCountryTemperatures
          |where AverageTemperature != '' and dt!=""
          |AND country='China'
          |and year(dt)>=1990
          |GROUP BY year(dt),season
          |ORDER BY year(dt),season
          | """.stripMargin)
    avgresult.show(false)
    DBTools.WriteMySql("china_season", avgresult)

  }

  //正则替换
  def regexpReplace(df: DataFrame, columnName: String, regexp: String, newValue: Any): DataFrame = {
    val exprString: String = "regexp_replace(" + columnName + ",'" + regexp + "','" + newValue + "')"
    df.withColumn(columnName, expr(exprString).alias(columnName))
  }

}
